package service

import (
	"fmt"
	"go-canal/config"
	"go-canal/internal"
	"go-canal/utils/database"
	"go-canal/utils/feishu"
	"go-canal/utils/kafka"
	"go-canal/utils/logger"
	"os"
	"os/signal"
)

var (
	AnalysisServer internal.CanalAnalysisServer
	Env            string
	Robot          *feishu.Robot
	s              chan os.Signal
)

func NewService(c config.Config) {
	Env = c.Env

	// 初始化日志
	logger.InitLogger(&c.LogOption)

	// 初始化数据库
	db := database.NewGorm(&c.Mysql)
	//redCli := redis.NewRedis(&c.Redis)

	// 初始化模块
	AnalysisServer = internal.NewCanalAnalysisServer(db)
	Robot = feishu.NewRobot(c.Webhook)

	// 一个topic设置一个消费者组,加快消费效率
	for i, topic := range c.CanalKafka.Topic {
		consumer := kafka.NewDefaultConsumer(c.CanalKafka.GroupID[i], c.CanalKafka.Addr, []string{topic}, CanalCustomerMessage)
		if err := consumer.Start(); err != nil {
			panic(err)
		}
	}

}

func Run() {
	fmt.Println("go-canal开始启动消费程序")
	s = make(chan os.Signal)
	signal.Notify(s, os.Interrupt)
	<-s
	stop()
}

func stop() {
	//在这里会等待所有的协程都退出
	fmt.Println("消费程序停止")
	os.Exit(1)
}
